package com.xuejiujiu.gmall.realtine.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange

import java.util
import scala.collection.mutable
import scala.util.parsing.json.JSON.headOptionTailToFunList

/**
 * Offset管理工具类，用于往redis中存储和读取Offset
 * 管理方案：
 * 1。后置提交偏移量  ->  手动控制偏移量的提交
 * 2. 手动控制偏移量提交  ->  SparkStreaming提供了手动的提交方案，但是我们不能用，因为我们会对DStream的结构进行转换
 * 3. 手动的提取偏移量维护到Redis中
 * -> 从Kafka中消费到数据，先提取偏移量
 * -> 等数据写入成功后，将偏移量存储到Redis中
 * -> 从Kafka中消费数据之前，先到Redis中读取偏移量，使用读取到的偏移量到Kafka中消费数据
 * 4. 手动的将偏移量存储到Redis中，每次消费数据，使用存储的Offset继续消费，每次消费到数据后要将本次消费到的Offset再存储到Redis
 */

object MyOffsetUtils {
  /**
   * 往Reids中存储Offset
   * 问题：
   * 存的Offset从哪来？
   *  从消费到的数据中提取出来，传入到该方法中
   *  offsetRanges: Array[OffsetRange]
   *
   * Offset的结构是什么？
   *  Kafka中offset维护的结构
   *  groupId + topic + partition => offset
   *  从传入进来的offset中提取关键信息
   *
   * 在redis中怎么存？
   * 类型： hash
   * key： gropuId + topic
   * value:partition - offset , partition - offset .....
   * 写入API： hset / hmset
   * 读取API: hgetall
   * 是否过期: 不过期
   *
   */
  def saveOffset(topic:String,groupId:String,offsetRanges: Array[OffsetRange]): Unit = {
    if(offsetRanges != null && offsetRanges.size >0){
      val offset = new util.HashMap[String, String]()
        for(offsetRange <- offsetRanges){
          val partition = offsetRange.partition
          val endOffset = offsetRange.untilOffset
          offset.put(partition.toString,endOffset.toString)
        }
      // 往redis中存
      val jedis = MyRedisUtils.getJedisFromPool()
      val redisKey = s"offset:$topic:$groupId"
      jedis.hset(redisKey,offset)
      jedis.close()
    }
  }

  /**
   * 从Redis中读取存储的Offset
   *
   * 问题：
   *  如何指定SparkStreaming通过指定的offset进行消费?
   *
   *  SparkStreaming要求的offset的格式是什么？
   *  Map[TopicPartition, Long]
   *
   */
  def readOffset(topic:String,groupId:String): Map[TopicPartition,Long] = {
    val jedis = MyRedisUtils.getJedisFromPool()
    val redisKey = s"offset:$topic:$groupId"
    val offsets: util.Map[String,String] = jedis.hgetAll(redisKey)
    val results = mutable.Map[TopicPartition, Long]()
    // 将Java的map转化为Scala的map进行迭代
    import scala.collection.JavaConverters._
    for ((partition,offset) <- offsets.asScala) {
      val tp = new TopicPartition(topic, partition.toInt)
      results.put(tp,offset.toLong)

    }
    jedis.close()
    results.toMap
  }

}
